package com.xb.system.service.impl;

import com.google.common.hash.BloomFilter;
import com.xb.model.domain.UsersInfo;
import com.xb.system.Filter.GlobalBloomFilter;
import com.xb.system.constants.BatchCSVConstants;
import com.xb.system.constants.ThreadPoolManager;
import com.xb.system.service.FileHandler;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVParser;
import org.apache.commons.csv.CSVRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.Reader;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;

/**
 * @author lzy
 * @create 2024/11/1 12:00
 */
@Component
@Slf4j
public class CSVDocumentHandler implements FileHandler {

    @Autowired
    private GlobalBloomFilter globalBloomFilter;  // 注入全局布隆过滤器

    @Autowired
    private DynamicDataSplitterServiceImpl splitterService;

    @Value("${file.batch.size}")
    private int batchSize;

    @Value("${file.thread.pool.size}")
    private int threadPoolSize;

    @Override
    public void handleFile(String file) throws IOException {
        BloomFilter<String> businessEmailBloomFilter = globalBloomFilter.getBusinessEmailBloomFilter();  // 使用全局布隆过滤器

        ThreadPoolManager.getExecutor(threadPoolSize);
        BatchCSVConstants.SIGNAL_COMPLETION.set(false);
        initConsumerThreads();
        BlockingQueue<UsersInfo> queue = BatchCSVConstants.QUEUE;

        try (InputStream is = Files.newInputStream(Paths.get(file));
             Reader reader = new InputStreamReader(is);
             CSVParser csvParser = CSVFormat.DEFAULT
                     .withHeader()  // 默认自动处理表头
                     .withIgnoreHeaderCase()
                     .withTrim()
                     .parse(reader)) {

            Iterator<CSVRecord> iterator = csvParser.iterator();
            if (iterator.hasNext()) {
                iterator.next(); // 跳过表头
            }

            while (iterator.hasNext()) {
                try {
                    CSVRecord record = iterator.next();
                    // 将 CSVRecord 转换为 String[]
                    String[] line = new String[record.size()];
                    for (int i = 0; i < record.size(); i++) {
                        line[i] = record.get(i);  // 将 CSVRecord 中的字段转为 String[]
                    }

                    // 处理每一行数据，使用全局布隆过滤器
                    UsersInfo userInfo = splitterService.processLine(line, businessEmailBloomFilter);
                    if (userInfo != null) {
                        queue.put(userInfo); // 将处理后的数据放入队列
                    }
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("处理 CSV 数据时发生中断: {}", e.getMessage());
                }
            }

        } catch (IOException e) {
            log.error("处理 CSV 文档时发生错误: {}", e.getMessage());
            throw e;
        }

        // 等待所有消费者线程完成处理
        BatchCSVConstants.SIGNAL_COMPLETION.set(true);
        ThreadPoolManager.shutdown();
        log.info("文件处理完成，线程池已关闭");
    }

    private void initConsumerThreads() {
        for (int i = 0; i < threadPoolSize; i++) {
            ThreadPoolManager.submitTask(() -> {
                try {
                    processQueue();
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    log.error("批处理线程被中断", e);
                }
            });
        }
    }

    private void processQueue() throws InterruptedException {
        batchSize = batchSize / 10;
        List<UsersInfo> batch = new ArrayList<>(batchSize);
        BlockingQueue<UsersInfo> queue = BatchCSVConstants.QUEUE;
        while (true) {
            UsersInfo userInfo = queue.poll(200, TimeUnit.MILLISECONDS);
            if (userInfo != null) {
                batch.add(userInfo);
            }

            if (batch.size() >= batchSize || (BatchCSVConstants.SIGNAL_COMPLETION.get() && queue.isEmpty())) {
                saveCSVBatch(batch);
                batch.clear();  // 清空当前批次
                if (BatchCSVConstants.SIGNAL_COMPLETION.get() && queue.isEmpty()) {
                    break;
                }
            }
        }
    }

    private void saveCSVBatch(List<UsersInfo> batch) {
        splitterService.saveBatch(batch);
    }
}
